package skewreduce.lsst.cook;

import skewreduce.framework.EaggDriver;
import skewreduce.framework.logical.LFinalizeOp;
import skewreduce.framework.logical.LMergeOp;
import skewreduce.framework.logical.LPartitionOp;
import skewreduce.framework.logical.LSampleOp;
import skewreduce.framework.logical.LProcessOp;
import skewreduce.framework.physical.PArrayPartitionOp;
import skewreduce.framework.physical.PPigOp;
import skewreduce.framework.physical.PhysicalOp;
import skewreduce.framework.physical.UniformSample;
import skewreduce.lib.Partition;
import skewreduce.lsst.Image2DPartition;

public class Cook extends EaggDriver {
	private int width;
	private int height;
	
    public String getAppName() { return "lsst.cook"; }

    public LSampleOp createLSampleOp() {
    	if ( getConf().getBoolean("lsst.cook.sample.uniform",true) )
    		return UniformSample.createLogicalOp(this);
    	throw new IllegalArgumentException();
//    	return new LSampleOp(this) {
//    		@Override
//    		protected PhysicalOp createInstance() {
//    			return new CookSample();
//    		}
//		};
    }
    
    public LPartitionOp createLPartitionOp() {
        return new LPartitionOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new PArrayPartitionOp();
            }
        };
    }

    public LProcessOp createLWorkOp() {
        return new LProcessOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new LocalCook();
            }
        };
    }

    public LMergeOp createLMergeOp() {
        return new LMergeOp(this) {
            @Override
            protected PhysicalOp createInstance() {
            	return new CookMerge();
            }
        };
    }

    public LFinalizeOp createLFinalizeOp() {
        return new LFinalizeOp(this) {
            @Override
            protected PhysicalOp createInstance() {
                return new PPigOp(); // FIXME
            }
        };
    }

    @Override
    protected int handleOption(String[] args,int i) {
        if ( "-numPart".equals(args[i]) ) {
            if ( i+1 < args.length ) {
                int n = Integer.parseInt(args[++i]);
                System.err.println("Overriding initial number of partitions to "+n);
                getConf().setInt("skewreduce.partition.initialpartitions",n);
                return 1;
            } else {
                System.err.println("missing -numPart parameter");
            }
        } else if ( "-width".equals(args[i]) ) {
        	width = Integer.parseInt(args[++i]);
            getConf().setInt("lsst.cook.width",width);
            return 1;
        } else if ( "-height".equals(args[i]) ) {
        	height = Integer.parseInt(args[++i]);
            getConf().setInt("lsst.cook.height",height);
        	return 1;
        } else if ( "-threshold".equals(args[i]) ) {
        	getConf().setFloat("lsst.cook.threshold",Float.parseFloat(args[++i]));
        	return 1;
        } else if ( "-buildAlways".equals(args[i]) ) {
            getConf().setBoolean("lsst.cook.buildAlways",true);
            return 1;
        }
        return -1;
    }

	@Override
	public Partition getRootPartition() {
		if ( width <= 0 || height <= 0 )
			throw new IllegalArgumentException(String.format("Invalid image size = %d x %d", width, height));
        getConf().set("skewreduce.array.dimension",String.format("%d,%d",width,height));
		// should generate from image size
		return new Image2DPartition(width,height);
	}
}
